package InvertedIndex;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.Random;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.StringTokenizer;


import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.map.InverseMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class InvertedIndex {

    public static class InvertedIndexMapper extends Mapper<LongWritable, Text, Text, Text> {

        private static Text keyInfo = new Text();// 存储单词和 URL 组合
        private static final Text valueInfo = new Text("1");// 存储词频,初始化为1
        private boolean caseSensitive;
        private Set<String> patternsToSkip = new HashSet<String>();
        private Configuration conf;
        private BufferedReader fis;
        private Set<String> wordstoskip = new HashSet<String>();

        public void setup(Context context) throws IOException,
                InterruptedException {
            conf = context.getConfiguration();
            caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
            if (conf.getBoolean("wordcount.skip.patterns", true)) {
                URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
                for (URI patternsURI : patternsURIs) {
                    if (patternsURI==patternsURIs[0]){
                        Path patternsPath = new Path(patternsURI.getPath());
                        String patternsFileName = patternsPath.getName().toString();
                        parseSkipFile(patternsFileName);
                    }
                }
            }
            if (conf.getBoolean("wordcount.skip.words", true)){
                URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
                for (URI patternsURI : patternsURIs) {
                    if (patternsURI!=patternsURIs[0]){
                        Path patternsPath = new Path(patternsURI.getPath());
                        String patternsFileName = patternsPath.getName().toString();
                        wordSkipFile(patternsFileName);
                    }
                }
            }
        }

        private void wordSkipFile(String filename){
            try {
                fis = new BufferedReader(new FileReader(filename));
                String pattern = null;
                while ((pattern = fis.readLine()) != null) {
                    wordstoskip.add(pattern);
                }
            } catch (IOException ioe) {
                System.err.println("Caught exception while parsing the cached file '"
                        + StringUtils.stringifyException(ioe));
            }
        }
        private void parseSkipFile(String fileName) {
            try {
                fis = new BufferedReader(new FileReader(fileName));
                String pattern = null;
                while ((pattern = fis.readLine()) != null) {
                    patternsToSkip.add(pattern);
                }
            } catch (IOException ioe) {
                System.err.println("Caught exception while parsing the cached file '"
                        + StringUtils.stringifyException(ioe));
            }
        }
        private static boolean isNumber(String str){

            String reg = "^[0-9]+(.[0-9]+)?$";

            return str.matches(reg);

        }

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            String line = (caseSensitive) ?
                    value.toString() : value.toString().toLowerCase();            String[] fields = StringUtils.split(line, ' ');// 得到字段数组
            for (String pattern : patternsToSkip) {
                line = line.replaceAll(pattern, "");
            }
            StringTokenizer itr = new StringTokenizer(line);
            FileSplit fileSplit = (FileSplit) context.getInputSplit();// 得到这行数据所在的文件切片
            String fileName = fileSplit.getPath().getName();// 根据文件切片得到文件名

            while (itr.hasMoreTokens()) {
                Text temp = new Text();
                temp.set(itr.nextToken());
                if (temp.getLength()>=3 && !wordstoskip.contains(temp.toString()) && !isNumber(temp.toString())){
                    keyInfo.set(itr.nextToken() + ":" + fileName);
                    context.write(keyInfo, valueInfo);
                }
            }

        }
    }

    public static class InvertedIndexCombiner extends Reducer<Text, Text, Text, Text> {

        private static Text info = new Text();

        // 输入： <MapReduce:file3 {1,1,...}>
        // 输出：<MapReduce file3:2>
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;// 统计词频
            for (Text value : values) {
                sum += Integer.parseInt(value.toString());
            }

            int splitIndex = key.toString().indexOf(":");
            // 重新设置 value 值由 URL 和词频组成
            info.set(key.toString().substring(splitIndex + 1) + ":" + sum);
            // 重新设置 key 值为单词
            key.set(key.toString().substring(0, splitIndex));

            context.write(key, info);
        }
    }



    public static class InvertedIndexReducer extends Reducer<Text, Text, Text, Text> {

        private static Text result = new Text();

        // 输入：<MapReduce file3:2>
        // 输出：<MapReduce file1:1;file2:1;file3:2;>
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            // 生成文档列表
            String fileList = new String();
            for (Text value : values) {
                fileList += value.toString() + ";";
            }

            result.set(fileList);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws IOException,
            ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
        String[] remainingArgs = optionParser.getRemainingArgs();
        if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
            System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
            System.exit(2);
        }
        Path tempDir = new Path("wordcount-temp-" + Integer.toString(
                new Random().nextInt(Integer.MAX_VALUE)));
        Job job = Job.getInstance(conf);

        job.setJarByClass(InvertedIndex.class);

        job.setMapperClass(InvertedIndexMapper.class);
        job.setCombinerClass(InvertedIndexCombiner.class);
        job.setReducerClass(InvertedIndexReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        List<String> otherArgs = new ArrayList<String>();
        for (int i=0; i < remainingArgs.length; ++i) {
            if ("-skip".equals(remainingArgs[i])) {
                job.addCacheFile(new Path(remainingArgs[++i]).toUri());
                job.getConfiguration().setBoolean("wordcount.skip.patterns", true);  //新增属性“wordcount.skip.patterns”,设置为true
            }
            else if ("-ignore".equals(remainingArgs[i])){
                job.addCacheFile(new Path(remainingArgs[++i]).toUri());
                job.getConfiguration().setBoolean("wordcount.skip.words", true);  //新增属性“wordcount.skip.patterns”,设置为true
            }
            else{
                otherArgs.add(remainingArgs[i]);
            }
        }
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        // 检查参数所指定的输出路径是否存在，若存在，先删除
        Path output = new Path(args[1]);
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(output)) {
            fs.delete(output, true);
        }
        FileOutputFormat.setOutputPath(job, output);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }


}